#pragma once

#include "comm/my_defs.h"
#include "comm/my_log.h"
#include "cache/util/my_cache_key.h"

#define MY_DISK_VOLUME_HEADER_TAG           0xe77ec5de
#define MY_DISK_VOLUME_LAYOUT_VERSION       1
#define MY_DISK_BLOCK_SIZE                  (4 * 1024 * 1024)       // 4M
#define MY_DISK_INDEX_ITEM_SIZE             12
#define MY_DISK_ONE_BUCKET_ITEMS            16                      // 16 items in one bucket
#define MY_DISK_ONE_BUCKET_MASK             (MY_DISK_ONE_BUCKET_ITEMS - 1)
#define MY_DISK_ONE_GROUP_BUCKETS           4096
#define MY_DISK_ONE_GROUP_ITEMS             (MY_DISK_ONE_GROUP_BUCKETS * MY_DISK_ONE_BUCKET_ITEMS)
#define MY_DISK_ONE_GROUP_SIZE              (MY_DISK_ONE_GROUP_ITEMS * MY_DISK_INDEX_ITEM_SIZE)
#define MY_DISK_MAX_VOLUME_SIZE             (48LL * 1024 * 1024 * 1024 * 1024) // 48T
#define MY_DISK_INDEX_TABLE_FILL_MULTIPLE   2   // index table fill factor is 0.5

//////////////// my_disk_volume_header ////////////////
typedef struct my_disk_volume_header_s my_disk_volume_header_t;

struct my_disk_volume_header_s {
    uint32_t            checksum;
    uint32_t            tag;
    uint32_t            version;
    uint32_t            block_size;
    int64_t             index_item_num;
    uint8_t             write_serial;
    uint8_t             recycle_serial;
    int64_t             write_pos;
    int64_t             recycle_pos;
    u_char              _pending[4048];  // aligned to page size
};

static inline void
my_disk_volume_header_init(my_disk_volume_header_t *header) {
    bzero(header, sizeof(my_disk_volume_header_t));
    header->tag = MY_DISK_VOLUME_HEADER_TAG;
    header->version = MY_DISK_VOLUME_LAYOUT_VERSION;
    header->block_size = MY_DISK_BLOCK_SIZE;
}

static inline void
my_disk_volume_header_update_checksum(my_disk_volume_header_t *header) {
    void *data = (void *)header;
    header->checksum = my_crc32_long(data + 4, sizeof(my_disk_volume_header_t) - 4);
}

/***
 * case1: recycle_serial == write_serial
 * |--invalid data--|recycle_pos--valid data--write_pos|--invalid data--|
 * 
 * case2: recycle_serial + 1 == write_serial
 * |--valid data--write_pos|--invalid data--|recycle_pos---valid data---|
 */
static inline my_bool_t
my_disk_serial_and_pos_vaild(uint8_t write_serial, int64_t write_pos,
    uint8_t recycle_serial, int64_t recycle_pos) {

    if (recycle_serial != write_serial && recycle_serial + 1 != recycle_serial) {
        MY_LOG_DEBUG("disk_layout", "serial number error, recycle_serial: %d, "
            "write_serial: %d", recycle_serial, write_serial);
        return MY_FALSE;
    }

    if (recycle_serial == write_serial) {
        if (write_pos < recycle_pos) {
            MY_LOG_DEBUG("disk_layout", "pos error in recycle_serial == write_serial, "
            "recycle_pos: %lld, write_pos: %lld", recycle_pos, write_pos);
            return MY_FALSE;
        }
        return MY_TRUE;
    }

    // recycle_serial + 1 == write_serial
    if (write_pos > recycle_pos) {
        MY_LOG_DEBUG("disk_layout", "pos error in recycle_serial + 1 = write_serial, "
            "recycle_pos: %lld, write_pos: %lld", recycle_pos, write_pos);
        return MY_FALSE;
    }

    return MY_TRUE;
}

static inline my_bool_t
my_disk_volume_header_check(my_disk_volume_header_t *header, int64_t index_item_num) {
    void *data = (void *)header;
    uint32_t checksum = my_crc32_long(data + 4, sizeof(my_disk_volume_header_t) - 4);
    if (header->checksum != checksum) {
        MY_LOG_WARN("disk_layout", "header checksum no match, checksum: %ld, "
            "header_checksum: %ld", checksum, header->checksum);
        return MY_FALSE;
    }

    if (header->tag != MY_DISK_VOLUME_HEADER_TAG
        || header->version != MY_DISK_VOLUME_LAYOUT_VERSION
        || header->block_size != MY_DISK_BLOCK_SIZE) {
        MY_LOG_WARN0("disk_layout", "tag or version no match");
        return MY_FALSE;
    }

    if (header->index_item_num != index_item_num) {
        MY_LOG_WARN0("disk_layout", "index_item_num no match");
        return MY_FALSE;
    }

    if (!my_disk_serial_and_pos_vaild(header->write_serial, header->write_pos,
        header->recycle_serial, header->recycle_pos)) {
        MY_LOG_WARN0("disk_layout", "serial and pos invalid");
        return MY_FALSE;
    }

    return MY_TRUE;
}


/////////////////////////// index table ///////////////////////////
static inline int64_t
my_disk_volume_index_table_size(int64_t index_item_num) {
    int64_t slot_num = index_item_num * MY_DISK_INDEX_TABLE_FILL_MULTIPLE;
    if (slot_num < MY_DISK_ONE_GROUP_ITEMS) {
        slot_num = MY_DISK_ONE_GROUP_ITEMS;
    }

    if (slot_num % MY_DISK_ONE_GROUP_ITEMS != 0) {
        slot_num = MY_ALIGN(slot_num, MY_DISK_ONE_GROUP_ITEMS);
    }

    return slot_num * MY_DISK_INDEX_ITEM_SIZE;
}


///////////////////////// index item ///////////////////////////

typedef struct my_disk_index_item_s my_disk_index_item_t;

struct my_disk_index_item_s {
    uint32_t        digest32:32;        // [0: 4):  digest64 to digest32
    unsigned int    block_id:24;        // [4: 7):  BLOCK_SIZE * 16M = 64T
    unsigned int    serial:8;           // [7: 8):  write serial
    unsigned int    off_block:16;       // [8: 10): offset in block. 64K * PAGE_SIZE = 256M
    unsigned int    size:16;            // [10:12): storage object agligned page size. 64K * PAGE_SIZE = 256M
};

static inline void
my_disk_index_item_init(my_disk_index_item_t *item, my_cache_key_t *cache_key,
    int64_t offset, int64_t size, uint8_t serial) {

    if (!MY_IS_PAGE_ALIGNED(offset) || !MY_IS_PAGE_ALIGNED(size)) {
        MY_LOG_FATAL("disk_layout", "offset: %d or size: %d not aligned page size.", offset, size);
    }

    int64_t block_id = offset / MY_DISK_BLOCK_SIZE;
    int64_t off_block = (offset % MY_DISK_BLOCK_SIZE) >> MY_PAGE_BITS;
    int64_t c_size = size >> MY_PAGE_BITS;

    item->digest32 = cache_key->digest32;
    item->block_id = block_id;
    item->off_block = off_block;
    item->size = c_size;
    item->serial = serial;
}

static inline int64_t
my_disk_index_item_offset(my_disk_index_item_t *item) {
    int64_t block_id = item->block_id;
    int64_t off_block = item->off_block;
    return (block_id * MY_DISK_BLOCK_SIZE) + (off_block << MY_PAGE_BITS);
}

static inline int64_t
my_disk_index_item_size(my_disk_index_item_t *item) {
    int64_t size = item->size;
    return size << MY_PAGE_BITS;
}

static inline my_bool_t
my_disk_check_env() {
    int64_t volume_header_size = sizeof(my_disk_volume_header_t);
    if (volume_header_size != MY_PAGE_SIZE) {
        MY_LOG_FATAL("disk_layout", "volume header size %lld not aligned page size",
            volume_header_size);
        return MY_FALSE;
    }

    int64_t index_item_size = sizeof(my_disk_index_item_t);
    if (index_item_size != MY_DISK_INDEX_ITEM_SIZE) {
        MY_LOG_FATAL("disk_layout", "index item size %lld no match", index_item_size);
        return MY_FALSE;
    }

    if (!MY_IS_PAGE_ALIGNED(MY_DISK_ONE_GROUP_SIZE)) {
        MY_LOG_FATAL("disk_layout", "group size %lld not aligned page size", 
            MY_DISK_ONE_GROUP_SIZE);
        return MY_FALSE;
    }

    return MY_TRUE;
}

////////////////////////////////// cache object ///////////////////////////////////
typedef struct {
    int32_t             key_len;
    int32_t             data_len;
    uint32_t            crc32;
    /////////////////////////////
    // key:  key_len           //
    // data: data_len          //
    /////////////////////////////
} my_disk_cache_object_t;